package com.heima.kafka.testDemo;

import org.apache.kafka.clients.producer.*;


import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class ProducerQuickStart {
    public static void main(String[] args) {
        //2.配置生产者的链接信息
        Properties properties = new Properties();
        //2.1配置kafka的连接地址信息
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.244.100:9092");
        //2.2配置kafka的key和value的序列化方式
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");

        //关于生产者的一些常见的参数配置

        //1.生产者消息确认机制,ack,三种值可选,0,1,all,分别表示不需要任何响应,需要leader接收并响应,需要所有的副本接收并响应
        properties.put(ProducerConfig.ACKS_CONFIG,"all");

        //2.生产者消息发送失败重试机制,设置重试次数,重试次数使用完毕才确认消息发送失败
        properties.put(ProducerConfig.RETRIES_CONFIG,10);

        //3.数据压缩,三种模式snappy,lz4,gzip,常用snappy
        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"snappy");


        //1.创建生产者对象
        KafkaProducer<String,String> producer = new KafkaProducer<String, String>(properties);
        //3.生产者发送消息
        //3.1封装发送的消息,指定topic,也可以指定分区partition
        ProducerRecord<String,String> record = new ProducerRecord<String,String>("kafka-first","key-001","hello,kafka");
        //3.2调用方法执行消息发送

        //同步发送形式
        /*try {
            RecordMetadata recordMetadata = producer.send(record).get();
            System.out.println(recordMetadata.offset());
        } catch (Exception e) {
            e.printStackTrace();
        }*/

        //消息异步发送
        producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                //判断当前有没有抛异常,可以在异常信息里记录日志,方便后期做补偿策略
                if (e != null){
                    System.out.println("记录异常信息到日志表中");
                }
                //打印偏移量
                System.out.println(recordMetadata.offset());
            }
        });
        //4.释放资源
        producer.close();
    }
}
